package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo14Student {
  def main(args: Array[String]): Unit = {
    //4、统计偏科最严重的前十学生

    //1、统计每个班级的总分的平均分

    // 创建spark环境
    val conf = new SparkConf()
    //local[*]： 使用计算所有的资源（CPU）
    conf.setMaster("local[*]")
    conf.setAppName("student")

    val sc = new SparkContext(conf)

    //读取分数表
    val scoreLinesRDD: RDD[String] = sc.textFile("spark/data/score.txt")

    //取出学号和分数
    val scoresRDD: RDD[(String, (String, Double))] = scoreLinesRDD
      .map(line => {
        val split: Array[String] = line.split(",")
        val id: String = split(0)
        val sId: String = split(1)
        val score: Double = split(2).toDouble
        (sId, (id, score))
      })

    //读取科目表
    val subjectLinesRDD: RDD[String] = sc.textFile("spark/data/subject.txt")
    val subjectRDD: RDD[(String, Double)] = subjectLinesRDD.map(line => {
      val split: Array[String] = line.split(",")
      val sid: String = split(0)
      val maxScore: Double = split(2).toDouble
      (sid, maxScore)
    })

    //关联分数表和科目表
    val joinRDD: RDD[(String, ((String, Double), Double))] = scoresRDD.join(subjectRDD)

    //整理数据,对分数进行归一化
    val idAndScoreRDD: RDD[(String, Double)] = joinRDD
      .map {
        case (_, ((id: String, score: Double), maxScore: Double)) =>
          (id, score / maxScore)
      }

    //按照学号进行分组
    val groupByRDD: RDD[(String, Iterable[Double])] = idAndScoreRDD.groupByKey()

    //计算每个学生分数的方差
    val stdRDD: RDD[(String, Double)] = groupByRDD
      .map {
        case (id: String, scores: Iterable[Double]) =>
          val num: Int = scores.size
          val sumScore: Double = scores.sum
          //计算平均数
          val avg: Double = sumScore / num
          //计算方差
          val std: Double = scores.map(score => (score - avg) * (score - avg)).sum / num
          (id, std)
      }

    //按照方差降序排序取前十
    val top10: Array[(String, Double)] = stdRDD
      .sortBy(kv => kv._2, ascending = false)
      .take(10)

    //取出前十学生的学号
    val ids: Array[String] = top10.map(kv => kv._1)

    //取出偏科最严重前十学生的分数
    val filterRDD: RDD[(String, (String, Double))] = scoresRDD
      .filter {
        case (sId: String, (id: String, score: Double)) =>
          //保留前十的学生
          ids.contains(id)
      }

    filterRDD.foreach(println)
  }
}
